package com.my.study.java.consumer;

import com.my.study.common.constant.TopicConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 消费者
 *
 * @author: yidujun
 * @create: 2021/07/23 10:53
 */
@Slf4j
public class ConsumerTest {

    private static final String BROKER_LIST = "47.98.50.37:9092";
    private static final String GROUP_ID = "group.demo";

    private static Properties initConfig() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer.id.demo");

        return props;
    }

    public static void main(String[] args) {
        AtomicBoolean isRunnable = new AtomicBoolean(true);
        Properties prop = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);

        // 订阅主题
        consumer.subscribe(Collections.singletonList(TopicConstants.TOPIC02));


        try {
            // 一直监听，循环消费消息
            while (isRunnable.get()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

                for (ConsumerRecord<String, String> record : records) {
                    if (null != record.value() || "" != record.value()) {
                        String value = record.value();
                        System.out.println(value);
                        System.out.println(record.toString());
                        System.out.println("=================================");
                        isRunnable.set(false);
                    }
                }
            }
        } catch (Exception e) {
            log.error("consumer error", e);
        } finally {
            consumer.close();
        }

    }
}
